package com.xxd.dfs.namenode;

import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-08-14 10:43
 * @description: 双层缓存区
 */
public class EditLogDoubleBuffer {
    private final static Logger logger = LoggerFactory.getLogger(EditLogDoubleBuffer.class);

    /** 默认缓冲区大小*/
    private int DEFAULT_BUFFER_SIZE = 0;
    /** editlog文件目录 */
    private String editLogDir = null;
    /** 第一层缓冲 */
    private EditLogBuffer bufCurrent = new EditLogBuffer();
    /** 刷新磁盘使用的缓冲 */
    private EditLogBuffer bufReady = new EditLogBuffer();
    /** 给backupnode同步使用的queue */
    private EditLogCache queue = EditLogCache.getInstance();

    private Configuration configuration = Configuration.getInstance();

    /** 维护edit log 的文件索引 */
    private List<String> editLogFiles = new CopyOnWriteArrayList<>();

    private long startTxid = 0L;

    private long endTxid = 0L;

    private long maxTxid = 0L;


    public EditLogDoubleBuffer() {
        DEFAULT_BUFFER_SIZE = Integer.valueOf(configuration.get("editlog.buffer.size"));
        editLogDir = configuration.get("editlog.dir");
    }

    /**
     * 获取所有文件
     * @return
     */
    public List<String> getEditLogFiles() {
        return editLogFiles;
    }

    /**
     * 设置所有的文件
     * @param editLogFiles
     */
    public void setEditLogFiles(List<String> editLogFiles) {
        this.editLogFiles.addAll(editLogFiles);
    }

    public void setStartTxid(long checkpointTxid) {
        this.startTxid = checkpointTxid + 1;
    }

    /**
     * 写入内存缓存
     *
     * @param editLog
     * @throws IOException
     */
    public void write(EditLog editLog) throws IOException {
        maxTxid = editLog.getTxid();
        String editLogJSON = JSONObject.toJSONString(editLog);

        queue.offer(editLog);

        bufCurrent.write(editLogJSON);
        bufCurrent.write("\n");
    }

    /**
     * 判断是否该刷入磁盘中去
     *
     * @return
     */
    public boolean shouldSyncToDisk() {
        return bufCurrent.size() >= DEFAULT_BUFFER_SIZE;
    }

    /**
     * 交换缓冲区
     */
    public void setReadyToFlush() {
        EditLogBuffer tmp = bufReady;
        bufReady = bufCurrent;
        bufCurrent = tmp;

        endTxid = maxTxid;
    }

    /**
     * 刷入磁盘
     */
    public void flushToDisk() throws IOException {
        bufReady.flush();
        bufReady.clear();
    }

    public void removeEditLogFile(String editLogFile) {
        this.editLogFiles.remove(editLogFile);
    }

    /**
     * 内存缓存
     */
    class EditLogBuffer {
        ByteArrayOutputStream buffer;

        public EditLogBuffer() {
            buffer = new ByteArrayOutputStream(DEFAULT_BUFFER_SIZE * 2);
        }

        public void write(String op) throws IOException {
            buffer.write(op.getBytes());
        }

        public int size() {
            return buffer.size();
        }

        /**
         * 讲数据刷入磁盘中
         * @throws IOException
         */
        public void flush() throws IOException {
            byte[] data = buffer.toByteArray();
            String editLogFile = "edits_" + startTxid + "-" + endTxid + ".log";
            String editLogFileAbsolute = editLogDir + editLogFile;

            startTxid = ++endTxid;
            editLogFiles.add(editLogFile);

            ByteBuffer buffer = ByteBuffer.wrap(data);

            RandomAccessFile file = null;
            FileOutputStream outputStream = null;
            FileChannel channel = null;

            try {
                file = new RandomAccessFile(editLogFileAbsolute, "rw");
                outputStream = new FileOutputStream(file.getFD());
                channel = outputStream.getChannel();

                channel.write(buffer);
                channel.force(false);
                logger.info("*******************磁盘刷入完毕************");
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (outputStream != null) {
                    outputStream.close();
                }
                if (file != null) {
                    file.close();
                }
                if (channel != null) {
                    channel.close();
                }
            }
        }

        /**
         * 清空内存缓冲里面的数据
         */
        public void clear() {
            buffer.reset();
        }
    }
}
